package com.ht.api.db.task;

import java.util.List;
import java.util.Map;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

import com.ht.api.db.Db;
import com.ht.api.db.Entity;
import com.ht.api.db.exception.HtException;
import com.ht.api.db.util.CronUtils;
import com.ht.api.db.util.HtUtil;
import com.ht.api.db.util.RedisUtil;

import cn.hutool.core.date.DateUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;

/**
 * 任务管理器
 * @author asq
 * @createTime 2024年12月9日 13:47:47
 */
@Slf4j
public class HtTaskExecutor {
	/** redis中存储所有任务组的hashset的key */
	public static final String KEY_GROUP = "task_group";
	/** redis中存储所有任务的hashset的key */
	public static final String KEY_TASK = "task";
	
	/** 最大线程数：CPU密集型 */
	//private final int cpuMaxThreadCount = Runtime.getRuntime().availableProcessors() + 1;
	/** 最大线程数：IO密集型 */
	private static final int ioMaxThreadCount = Runtime.getRuntime().availableProcessors() * 2;
	/** 调度器 */
	private static final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(ioMaxThreadCount);

	/** 任务存储容器 */
	private static final Map<Long, TaskInfo> tasks = new ConcurrentHashMap<>();
	/** 任务组存储容器 */
	public static final Map<String, TaskGroup> groups = new ConcurrentHashMap<>();
	/** 操作延迟10毫秒 */
	private static final int OPERATE_DELAY_TIME = 0;
	
	/**
	 * 立即执行多个普通任务，并阻塞等待所有任务都执行完后返回（默认构造10个线程）
	 * @param taskList 任务列表
	 */
	public static void execute(List<Runnable> taskList) {
		execute(10, taskList);
	}
	
	/**
	 * 立即执行多个普通任务，并阻塞等待所有任务都执行完后返回
	 * @param nThreads 线程池数量
	 * @param taskList 任务列表
	 */
	public static void execute(int nThreads, List<Runnable> taskList) {
	    ExecutorService executor = Executors.newFixedThreadPool(nThreads);
	    CountDownLatch latch = new CountDownLatch(taskList.size());
	    
	    try {
	        for (Runnable runnable : taskList) {
	            executor.submit(() -> {
	                try {
	                    runnable.run();
	                } finally {
	                    latch.countDown();
	                }
	            });
	        }

	        latch.await();
	    } catch (InterruptedException e) {
	        log.error("多线程执行多个任务，等待所有任务完成时发生异常：{}", e.getMessage(), e);
	        Thread.currentThread().interrupt(); // 恢复中断状态
	    } finally {
	        executor.shutdown();
	        try {
	            if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
	                executor.shutdownNow();
	            }
	        } catch (InterruptedException e) {
	            executor.shutdownNow();
	            Thread.currentThread().interrupt();
	        }
	    }
	}

	/**
	 * 立即执行一个普通任务
	 * @param task 任务
	 */
	public static void execute(TimerTask task) {
		scheduler.schedule(task, OPERATE_DELAY_TIME, TimeUnit.MILLISECONDS);
	}
	
	/**
	 * 立即执行一个普通任务
	 * @param runnable 任务
	 */
	public static void execute(Runnable runnable) {
	    execute(new TimerTask() {
	        @Override
	        public void run() {
	            runnable.run();
	        }
	    });
	}
	
	/**
	 * 提交一个任务，立即执行一次后结束
	 * @param requestParams
	 * @param handler
	 * @return
	 */
	public static long submitTask(JSONObject requestParams, TaskHandler handler) {
		return submitTask(null, requestParams, handler, null, handler.getClass().getSimpleName());
	}
	
	/**
	 * 提交一个任务，立即执行一次后结束
	 * @param requestParams
	 * @param handler
	 * @param name 任务具体名称
	 * @return
	 */
	public static long submitTask(JSONObject requestParams, TaskHandler handler, String name) {
		return submitTask(null, requestParams, handler, null, name);
	}
	
	/**
	 * 提交一个任务（指定任务组），立即执行一次后结束
	 * @param requestParams
	 * @param handler
	 * @return
	 */
	public static long submitTask(TaskGroup group, JSONObject requestParams, TaskHandler handler) {
		return submitTask(group, requestParams, handler, null, handler.getClass().getSimpleName());
	}
	
	/**
	 * 提交一个任务（指定任务组），立即执行一次后结束
	 * @param requestParams
	 * @param handler
	 * @param name 任务具体名称
	 * @return
	 */
	public static long submitTask(TaskGroup group, JSONObject requestParams, TaskHandler handler, String name) {
		return submitTask(group, requestParams, handler, null, name);
	}
	
	/**
	 * 提交一个任务，并指定指行计划
	 * @param requestParams
	 * @param handler
	 * @param plan
	 * @return
	 */
	public static long submitTask(JSONObject requestParams, TaskHandler handler, Plan plan) {
		return submitTask(null, requestParams, handler, plan, handler.getClass().getSimpleName());
	}
	
	/**
	 * 提交一个任务，并指定指行计划
	 * @param requestParams
	 * @param handler
	 * @param plan
	 * @param name 任务具体名称
	 * @return
	 */
	public static long submitTask(JSONObject requestParams, TaskHandler handler, Plan plan, String name) {
		return submitTask(null, requestParams, handler, plan, name);
	}

	
	/**
	 * 提交一个任务（指定任务组），并指定指行计划
	 * @param group 任务组
	 * @param requestParams 任务参数
	 * @param handler 任务执行器
	 * @param plan 执行计划
	 * @return
	 */
	public static long submitTask(TaskGroup group, JSONObject requestParams, TaskHandler handler, Plan plan) {
		return submitTask(group, requestParams, handler, plan, handler.getClass().getSimpleName());
	}
	
	/**
	 * 提交一个任务（指定任务组），并指定指行计划
	 * @param group 任务组
	 * @param requestParams 任务参数
	 * @param handler 任务执行器
	 * @param plan 执行计划
	 * @param name 任务具体名称
	 * @return
	 */
	public static long submitTask(TaskGroup group, JSONObject requestParams, TaskHandler handler, Plan plan, String name) {
		// 创建任务信息
		String now = DateUtil.now();
		TaskInfo task = TaskInfo.builder().taskId(HtUtil.getIdLong()).requestParams(requestParams).createTime(now)
				.startTime(now).executeOnce(true).status(TaskInfo.RUNNING).handler(handler)
				.baseName(handler.getName()).name(name).build();
		
		// 设置任务的执行计划
		if (plan != null) {
			plan.setPlan(task);
		}
		
		// 存储任务组信息
		if (group != null) {
			if (!groups.containsKey(group.getId())) {
				groups.put(group.getId(), group);
				
				// 任务组写入数据库
				//Db.insert(group.toEntry());
				save(group.toEntry(), true, false);
			}
			
			// 任务组关联任务
			task.setGroupId(group.getId());
		}
		
		// 任务写入数据库
		save(task.toEntry(), false, false);

		// 开始任务
		return startTask(task);
	}
	
	/**
	 * 根据任务ID开始一个任务
	 * @param taskId
	 */
	public static void startTask(String taskId) {
		Entity task = RedisUtil.hget(HtTaskExecutor.KEY_TASK, taskId);
		if (task == null) {
			task = Db.createSqlParser("db_task").findById(taskId);
		}
		if (task == null) {
			throw new HtException("任务不存在，taskId:" + taskId);
		}
		
		// 加载任务
		TaskInfo taskInfo = TaskInfo.load(task);

		// 恢复标记
		taskInfo.setAttempts(0);
		taskInfo.setCanceled(false);
		
		// 存储任务
		tasks.put(taskInfo.getTaskId(), taskInfo);
		save(taskInfo.toEntry(), false, false);
		
		// 立即开始执行
		scheduleTask(taskInfo, 0);
	}

	/**
	 * 开始执行一个任务
	 * @param task 任务
	 * @return
	 */
	public static long startTask(TaskInfo task) {
		// 恢复标记
		task.setCanceled(false);
		
		// 存储任务
		tasks.put(task.getTaskId(), task);
		save(task.toEntry(), false, false);

		// 开始执行
		if (task.getAttempts() == 0) {
			long initialDelay = calculateInitialDelay(task);
			if (initialDelay > 0) {
				scheduleTask(task, initialDelay);
			} else {
				scheduleTask(task, calculateNextDelay(task));
			}
		} else {
			scheduleTask(task, calculateNextDelay(task));
		}

		return task.getTaskId();
	}

	/**
	 * 停止任务
	 */
	public static void stopTask(long taskId) {
		TaskInfo task = tasks.get(taskId);
		if (task != null && task.getFuture() != null) {
			// 标记为需要取消任务
			task.setCanceled(true);
			
			// 取消任务
			task.getFuture().cancel(false);
		} else {
			log.warn("[task]未找到任务，无法停止，taskId：{}", taskId);
		}
	}
	
	/**
	 * 修改执行计划
	 * @param taskId 任务ID
	 * @param plan 新的执行计划
	 */
	public static TaskInfo changePlan(long taskId, Plan plan) {
		// 查询任务，先从缓存中获取，缓存中没有时，再查询数据库
		boolean isRunning = true;
		TaskInfo task = tasks.get(taskId);
		if (task == null) {
			isRunning = false;
			Entity dbTask = Db.createSqlParser("db_task").findById(taskId);
			if (dbTask == null) {
				log.warn("修改任务执行计划失败，taskId不存在：{}", taskId);
				return null;
			}
			task = TaskInfo.load(dbTask);
		}
		
		// 如果任务在运行中，则先停止任务
		if (isRunning) {
			HtTaskExecutor.stopTask(taskId);
		}
		
		// 修改执行计划
		plan.setPlan(task);
		task.setAttempts(0); // 将当前执行计划下的执行次数归0，再次执行任务时，便可以从头开始执行
		save(task.toEntry(), false, !isRunning);
		log.info("修改任务执行计划，taskId:{}", taskId);
		
		// 如果任务在运行中，则恢复启动任务
		if (isRunning) {
			HtTaskExecutor.startTask(task);
		}
		
		return task;
	}

	/**
	 * 获取任务信息
	 */
	public static TaskInfo getTaskInfo(long taskId) {
		return tasks.get(taskId);
	}

	/**
	 * 关闭管理器
	 */
	public static void shutdown() {
		scheduler.shutdown();
		try {
			if (!scheduler.awaitTermination(60, TimeUnit.SECONDS)) {
				scheduler.shutdownNow();
			}
		} catch (InterruptedException e) {
			scheduler.shutdownNow();
			Thread.currentThread().interrupt();
		}
	}

	/**
	 * 获取系统当前时间戳（秒级）
	 * @return
	 */
	private static int getCurrentTime() {
		return (int) (System.currentTimeMillis() / 1000);
	}
	
	/**
	 * 使用指定延迟调度任务
	 */
	private static void scheduleTask(TaskInfo task, long initialDelay) {
		//cancelExistingTask(task);
		
		ScheduledFuture<?> future = scheduler.schedule(() -> {
			try {
				executeTask(task);
			} catch (Exception e) {
				log.error("Task execution failed: {}", task.getTaskId(), e);
				handleTaskError(task, e);
			}
		}, initialDelay, TimeUnit.SECONDS);

		task.setFuture(future);
	}
	
	/**
	 * 执行任务
	 */
	private static void executeTask(TaskInfo task) {
		try {
			// 执行任务
			log.info("[task]任务池：{}，第{}次执行任务：{}", tasks.size(), task.getAttempts() + 1, 
					JSONUtil.toJsonStr(task.getRequestParams()));
			
			// 执行任务
			if (task.getHandler() != null) {
				// 执行次数+1
				task.setAttempts(task.getAttempts() + 1); // 当前计划下的执行次数+1
				task.setTotalAttempts(task.getTotalAttempts() + 1); // 累计执行次数+1
				task.setLastExecuteTime(DateUtil.now());
				
				// 执行
				boolean isSuccess = task.getHandler().handler(task.getRequestParams(), task.getGroupId(), task.getTaskId());
				if (!isSuccess) {
					task.setFailAttempts((task.getFailAttempts() + 1)); // 失败次数+1
				}
			} else {
				log.error("[task]未指定任务执行器，无法执行任务，任务ID:{}", task.getTaskId());
				completeTask(task, TaskInfo.FAILED, "没有任务执行器");
				return;
			}
			
			// 更新任务信息
			task.setStatus(TaskInfo.RUNNING);
			save(task.toEntry(), false, false);
			
			// 如果任务已经被外部取消，则终止
			if (task.isCanceled()) {
				completeTask(task, TaskInfo.STOPPED, "已手动停止");
				log.info("[task]已取消任务，taskId：{}", task.getTaskId());
				return;
			}

			// 没有cron表达式时，才进行终止条件检测
			if (task.getCronExpression() == null) {
				// 检查是否只执行1次
				if (task.isExecuteOnce()) {
					completeTask(task, TaskInfo.COMPLETED, "已完成一次执行");
					return;
				}
				
				// 检查是否超出最大执行次数
				if (task.getMaxNum() > 0 && task.getAttempts() >= task.getMaxNum()) {
					completeTask(task, TaskInfo.COMPLETED, "已完成规定的最大执行次数");
					return;
				}
	
				// 检查是否超出最大执行时间
				if (isTimeout(task)) {
					completeTask(task, TaskInfo.COMPLETED, "已完成规定的最大执行时间");
					return;
				}
			}
			
			// 继续执行
			scheduleTask(task, calculateNextDelay(task));
		} catch (Exception e) {
			task.setFailAttempts((task.getFailAttempts() + 1)); // 失败次数+1
			completeTask(task, TaskInfo.FAILED, e.getMessage());
			throw new RuntimeException("执行任务失败", e);
		}
	}

	/**
	 * 计算初始延迟时间
	 */
	private static long calculateInitialDelay(TaskInfo task) {
	    // 如果有cron表达式，优先使用cron表达式计算延迟
	    if (task.getCronExpression() != null) {
	        return CronUtils.getDelaySeconds(task.getCronExpression());
	    }
		
		// 按执行计划的配置，计算初始执行时间点
		int currentTime = getCurrentTime();
		long startTime = task.toTimestamp(task.getStartTime());
		if (startTime > currentTime) {
			return startTime - currentTime;
		}
		return 0;
	}

	/**
	 * 计算下次执行延迟
	 */
	private static long calculateNextDelay(TaskInfo task) {
	    // 如果有cron表达式，计算表达式的下次执行时间
		long cronNextDelay = Long.MAX_VALUE;
	    if (task.getCronExpression() != null) {
	    	cronNextDelay = CronUtils.getDelaySeconds(task.getCronExpression());
	    }
		
	    // 如果本轮任务已结束，则直接返回cron的下次时间点
	    if (task.isOver()) {
	    	task.setAttempts(0);	    	
	    	return cronNextDelay;
	    }
	    
	    // 按执行计划的配置，计算下次时间点
	    long planNextDelay = Long.MAX_VALUE;
		if (task.getIntervalMultiplier() <= 0) {
			planNextDelay = task.getInitialInterval();
		} else {
			int attempts = Math.max(task.getAttempts() - 1, 0);
			long delay = (long) (task.getInitialInterval() * Math.pow(task.getIntervalMultiplier(), attempts));
			planNextDelay = Math.min(delay, task.getMaxInterval());
		}
		
		// cron的下次时间和执行计划的下次时间，取最近的一个
		if (cronNextDelay < planNextDelay) {
	    	task.setAttempts(0);
			return cronNextDelay;
		} else {
			return planNextDelay;
		}
	}

	/**
	 * 检查是否已达到允许的最大执行时间
	 */
	private static boolean isTimeout(TaskInfo task) {
		// 注意：如果maxExecutionHours参数未设置，将永久执行
		if (task.getMaxTimes() <= 0) {
			return false;
		}
		long executionTimeInSeconds = getCurrentTime() - task.toTimestamp(task.getCreateTime());
		return executionTimeInSeconds >= task.getMaxTimes();
	}

	/**
	 * 完成任务
	 */
	private static void completeTask(TaskInfo task, String status, String message) {
		task.setStatus(status);
		task.setMessage(message);
		tasks.remove(task.getTaskId());

		// 写入数据库
		save(task.toEntry(), false, true);
		
		// 如果任务有任务组，则将任务组的任务完成数+1
		if (task.getGroupId() != null) {
			TaskGroup taskGroup = groups.get(task.getGroupId());
			if (taskGroup != null) {
				boolean isOver = taskGroup.markDone();
				
				// 将任务组信息写入数据库
				save(taskGroup.toEntry(), true, isOver);
				
				// 如果任务组的全部任务已完成，则从存储容器中移除
				if (isOver) {
					groups.remove(task.getGroupId());
				}
			}
		}
	}

	/**
	 * 处理任务错误
	 */
	private static void handleTaskError(TaskInfo task, Exception e) {
		task.setMessage(e.getMessage());
		if (!task.isExecuteOnce()) {
			scheduleTask(task, calculateNextDelay(task));
		}
	}
	
	/**
	 * 保存任务或任务组
	 * @param entity 要保存的任务或任务组
	 * @param isOver 是否是任务组
	 * @param isOver 是否已完成任务
	 */
	private static void save(Entity entity, boolean isGroup, boolean isOver) {
		// 获取缓存key
		String key = isGroup ? KEY_GROUP : KEY_TASK;
		
		// 存缓存
		RedisUtil.hset(key, entity.getMainId(), entity);
		
		// 如果任务已真正完成，则清除缓存并写入数据库
		if (isOver) {
			execute(() -> {
				try {
					// 真正存数据库
					Db.insert(entity);
				} finally {
					// 清除缓存
					RedisUtil.delete(key, entity.getMainId());
				}
			});
		}
	}
}